package com.stefanie.kafkademo.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author: stefanie-zy
 * @time: 2023/12/25-16:35
 * @description: kafka-生产者拦截器
 */
@Slf4j
public class MyProducerInterceptor implements ProducerInterceptor<String, String> {

    private static long sendSuccessNum = 0;
    private static long sendFailNum = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        // 重新定义消息前缀[消息体新增'stefanie-']
        String modifiedKeyValue = "stefanie-" + producerRecord.value();
        return new ProducerRecord<>(producerRecord.topic(), producerRecord.partition(), producerRecord.timestamp(), producerRecord.key(), modifiedKeyValue, producerRecord.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            sendSuccessNum++;
        } else {
            sendFailNum++;
        }
    }

    @Override
    public void close() {
        double sendSuccessRatio = (double) sendSuccessNum / (sendFailNum + sendSuccessNum);
        log.info("消息发送成功率：{}", String.format("%f", sendSuccessRatio * 100) + "%");
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
